package com.zhg.shortlink.service;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import com.zhg.shortlink.constant.RedisKeyConstants;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author 朱洪刚
 * @version 1.0
 * @data 2025/10/21 09:25
 */
@Slf4j
@RequiredArgsConstructor
@Service
@Deprecated
public class LocalBloomFilterService {
    private final RedisTemplate<String, String> redisTemplate;

    @Value("${shortlink.bloom.expected-insertions:10000000}")
    private int expectedInsertions;

    @Value("${shortlink.bloom.false-probability:0.01}")
    private double falseProbability;

    @Value("${shortlink.redis.fetchDataSize:5000}")
    private int fetchDataSize;

    private BloomFilter<String> localBloomFilter;
    private final ConcurrentLinkedQueue<String> pendingSync = new ConcurrentLinkedQueue<>();
    private final AtomicLong syncCounter = new AtomicLong(0);

    private BloomFilterStreamService streamService; // 延迟注入，避免循环依赖

    // 设置Stream服务（避免循环依赖）
    public void setStreamService(BloomFilterStreamService streamService) {
        this.streamService = streamService;
    }
    @PostConstruct
    public void initLocalBloomFilter() {
        localBloomFilter = BloomFilter.create(
                Funnels.stringFunnel(Charset.defaultCharset()),
                expectedInsertions,
                falseProbability
        );

        // 从Redis加载已有数据到本地布隆过滤器
        loadFromRedis();

        log.info("本地布隆过滤器初始化完成");
    }

    public boolean mightContain(String shortCode) {
        return localBloomFilter.mightContain(shortCode);
    }

    /**
     * 添加短链（会触发Stream同步到其他节点）
     */
    public void add(String shortCode) {
        localBloomFilter.put(shortCode);
        pendingSync.offer(shortCode); // 加入待同步队列

        // 发布到Stream，同步到其他节点
        if (streamService != null) {
            streamService.publishNewShortCode(shortCode);
        }
    }

    /**
     * 仅添加到本地（不触发Stream发布，用于接收同步消息）
     */
    public void addLocal(String shortCode) {
        localBloomFilter.put(shortCode);
        log.debug("本地添加短链到布隆过滤器: {}", shortCode);
    }

    /**
     * 批量添加（会触发Stream同步）
     */
    public void addBatch(List<String> shortCodes) {
        for (String shortCode : shortCodes) {
            localBloomFilter.put(shortCode);
            pendingSync.offer(shortCode);
        }

        // 批量发布到Stream
        if (streamService != null) {
            streamService.publishBatchShortCodes(shortCodes);
        }

        log.info("批量添加{}个短链到本地布隆过滤器", shortCodes.size());
    }

    @Scheduled(fixedRate = 5000)
    public void syncToRedis() {
        if (pendingSync.isEmpty()) {
            return;
        }

        try {
            int batchSize = Math.min(fetchDataSize, pendingSync.size());
            String[] batch = new String[batchSize];

            for (int i = 0; i < batchSize && !pendingSync.isEmpty(); i++) {
                batch[i] = pendingSync.poll();
            }

            if (batch.length > 0) {
                redisTemplate.opsForSet().add(RedisKeyConstants.BLOOM_SET_KEY, batch);
                syncCounter.addAndGet(batch.length);
                log.debug("同步{}个元素到Redis", batch.length);
            }
        } catch (Exception e) {
            log.error("同步到Redis失败", e);
        }
    }

    private void loadFromRedis() {
        try {
            // 直接从主数据源加载，简化逻辑
            long count = loadFromRedisWithScan(RedisKeyConstants.BLOOM_SET_KEY);
            if (count > 0) {
                log.info("从Redis加载{}个元素到本地布隆过滤器", count);
            } else {
                log.info("Redis中暂无布隆过滤器数据");
            }
        } catch (Exception e) {
            log.error("从Redis加载数据失败", e);
        }
    }

    /**
     * 使用SCAN命令分批加载数据
     */
    private long loadFromRedisWithScan(String key) {
        long totalCount = 0;

        try {
            // 使用SCAN命令遍历Set
            ScanOptions scanOptions = ScanOptions.scanOptions()
                    .match("*")
                    .count(fetchDataSize)
                    .build();

            Cursor<String> cursor = redisTemplate.opsForSet().scan(key, scanOptions);

            List<String> batch = new ArrayList<>();
            while (cursor.hasNext()) {
                batch.add(cursor.next());

                // 达到批次大小时处理一批
                if (batch.size() >= fetchDataSize) {
                    processBatch(batch);
                    totalCount += batch.size();
                    batch.clear();

                    // 避免长时间占用CPU，让出执行权
                    Thread.yield();
                }
            }

            // 处理最后一批
            if (!batch.isEmpty()) {
                processBatch(batch);
                totalCount += batch.size();
            }

            cursor.close();

        } catch (Exception e) {
            log.error("分批加载Redis数据失败: key={}", key, e);
        }

        return totalCount;
    }

    /**
     * 处理一批数据
     */
    private void processBatch(List<String> batch) {
        for (String shortCode : batch) {
            localBloomFilter.put(shortCode);
        }
        log.debug("处理一批数据: {} 条", batch.size());
    }

    /**
     * 异步同步数据到本地key
     */
    private void asyncSyncToLocalKey(String sourceKey, String targetKey) {
        CompletableFuture.runAsync(() -> {
            try {
                // 分批复制数据
                ScanOptions scanOptions = ScanOptions.scanOptions()
                        .match("*")
                        .count(1000)
                        .build();

                Cursor<String> cursor = redisTemplate.opsForSet().scan(sourceKey, scanOptions);
                List<String> batch = new ArrayList<>();

                while (cursor.hasNext()) {
                    batch.add(cursor.next());

                    if (batch.size() >= 1000) {
                        redisTemplate.opsForSet().add(targetKey, batch.toArray(new String[0]));
                        batch.clear();
                    }
                }

                if (!batch.isEmpty()) {
                    redisTemplate.opsForSet().add(targetKey, batch.toArray(new String[0]));
                }

                cursor.close();
                log.info("异步同步数据到本地key完成");

            } catch (Exception e) {
                log.error("异步同步数据失败", e);
            }
        });
    }

    public String getStats() {
        return String.format("本地布隆过滤器 - 预期容量: %d, 误判率: %.4f, 已同步: %d, 待同步: %d",
                expectedInsertions, falseProbability, syncCounter.get(), pendingSync.size());
    }
}
